package cn.itcast.flink.process;

import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Iterator;

/**
 * Author itcast
 * Date 2022/1/17 15:58
 * Desc 在服务器运维中，需要实时监控服务器机架的温度，如果一定时间内温度超过了一定阈值(100度)，
 * 且后一次上报的温度超过了前一次上报的温度，需要触发告警（温度持续升高中）
 * 使用 onTimer 触发器，通过给定的时间，触发事件
 * 超过 5s 触发触发器， timeout(): 触发告警信息 （发邮件、打电话等）
 */
public class SystemMonitorDemo {
    public static void main(String[] args) throws Exception {
        //初始化流计算运行环境，指定并行度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //接入socket数据源，获取数据  1,100
        DataStreamSource<String> source = env.socketTextStream("node1", 9999);
        //将获取到的数据转换成tuple2<Integer,Integer>
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = source.map(new MapFunction<String, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> map(String value) throws Exception {
                String[] arrs = value.split(",");
                return Tuple2.of(
                        Integer.parseInt(arrs[0]),
                        Integer.parseInt(arrs[1])
                );
            }
        })
                //根据 f0进行分流, keyedStream 才有 onTimer 触发器
                .keyBy(t -> t.f0);

        keyedStream.process(new MyKeyedProcessFunction())
                //打印输出
                .print();
        //执行任务
        env.execute();
    }
    //自定义ProcessFunction对象，继承KeyedProcessFunction<Tuple, Tuple2<Integer, Integer>, String>抽象类

    /**
     * KeyedProcessFunction<Integer ,Tuple2<Integer,Integer>,String>
     * <K> Type of the key.
     * <I> Type of the input elements.
     * <O> Type of the output elements.
     */
    public static class MyKeyedProcessFunction extends KeyedProcessFunction<Integer, Tuple2<Integer, Integer>, String> {
        //初始化 ListState<Tuple2<机架id, 机架温度>> 保存上次温度
        ListState<Tuple2<Integer, Integer>> lastTemperatureState = null;
        //定义触发警告定时器的时长和格式化为"yyyy-MM-dd HH:mm:ss.SSS"

        /**
         * 做初始化的工作，获取上一次的状态
         *
         * @param parameters
         * @throws Exception
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            //获取ListState Tuple2<Integer, Integer>
            lastTemperatureState = getRuntimeContext().getListState(
                    new ListStateDescriptor(
                            "lastTemperatureState",
                            Types.TUPLE(Types.INT, Types.INT)
                    )
            );
        }

        /**
         * 来一条数据处理一条数据
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement(Tuple2<Integer, Integer> value, Context ctx, Collector<String> out) throws Exception {
            Integer lastTemprature = 0;
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            //定时器触发时间
            long timerTime = 0L;
            //将listState数据赋值给最后一次温度数据，如果没有赋值为Tuple2<0,0>
            Iterable<Tuple2<Integer, Integer>> temperatures = lastTemperatureState.get();
            for (Tuple2<Integer, Integer> lastTemperature : temperatures) {
                lastTemprature = lastTemperature.f1;
            }
            //如果温度>100并且大于状态温度，将数据保存到listState中，并注册个定时器，5s触发告警，收集温度高告警信息
            if (value.f1 > 100 && value.f1 > lastTemprature) {
                //如果超过100度并且大于上次的温度，将当前的温度保存到 state 中
                lastTemperatureState.add(value);
                //ctx.timerService().currentProcessingTime()+ 5000L;
                long timestamp = ctx.timerService().currentProcessingTime();
                timerTime = timestamp + 10000;
                //通过上下文注册处理时间的触发器
                ctx.timerService().registerProcessingTimeTimer(timerTime);
                out.collect("当前的温度是 " + value.f1 + "当前的时间是：" + sdf.format(timestamp));
            } else { //如果温度下来了，删除触发器时间
                //温度下来了，不需要再报警
                //否则 清空状态温度，并取消定时器 ctx.timerService().deleteProcessingTimeTimer(timeTS);
                lastTemperatureState.clear();
                //删除触发器
                ctx.timerService().deleteProcessingTimeTimer(timerTime);
                out.collect("当前解除告警！");

            }
        }

        /**
         * 定时到期触发的事件
         *
         * @param timestamp
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //获取状态中数据size
            Iterator<Tuple2<Integer, Integer>> iterator = lastTemperatureState.get().iterator();
            int size = IteratorUtils.toList(iterator).size();
            System.out.println("当前状态中的超过100度并且持续增长的温度个数为：" + size + " 个！");
            //如果大于 1 触发告警，并收集告警信息
            if (size > 1) {
                System.out.println("触发告警了！");
                out.collect("触发告警了！");
                //调用外部接口，帮助实现告警信息 controller("/A/B/C/email/call")
            }
            //清空历史数据
            lastTemperatureState.clear();
        }
    }

}
